package org.ala.tiktools.rpc.netty.handler;

import java.util.concurrent.CompletableFuture;

import org.ala.tiktools.logs.LoggerFactory;
import org.ala.tiktools.rpc.client_server.service.RpcServiceExecutor;
import org.ala.tiktools.rpc.config.RpcConfigInfo;
import org.ala.tiktools.rpc.exception.RpcLoginException;
import org.ala.tiktools.rpc.exception.RpcNoLoginException;
import org.ala.tiktools.rpc.exception.RpcReadException;
import org.ala.tiktools.rpc.exception.RpcRemoteException;
import org.ala.tiktools.rpc.exception.RpcWriteException;
import org.ala.tiktools.rpc.msg.RpcMsg;
import org.ala.tiktools.rpc.msg.RpcMsgHead;
import org.ala.tiktools.rpc.msg.ack.RpcAckBody;
import org.ala.tiktools.rpc.msg.ack.RpcAckMsg;
import org.ala.tiktools.rpc.msg.pong.RpcPongMsg;
import org.ala.tiktools.rpc.msg.push.RpcPushBody;
import org.ala.tiktools.rpc.msg.push.RpcPushMsg;
import org.ala.tiktools.rpc.netty.attr.RpcAttribute;
import org.ala.tiktools.rpc.plugin.serializer.SerializeThreadPool;
import org.ala.tiktools.rpc.plugin.serializer.kryo.KryoThreadSerializer;
import org.slf4j.Logger;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * 读到消息处理器
 *
 * @author ala
 * @date 2025-04-08 12:55
 */
public class RpcMsgReadHandler extends ChannelInboundHandlerAdapter {


    static Logger log = LoggerFactory.getLogger("rpc", "handler");


    /**
     *  业务执行器
     *  <p> 对客户端而言，这个属性为null
     *  <p> 执行器本身就是异步执行的
     */
    protected RpcServiceExecutor serviceExecutor;


    public RpcMsgReadHandler(RpcServiceExecutor se) {
        this.serviceExecutor = se;
    }


    @SuppressWarnings({ "rawtypes", "incomplete-switch" })
	@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        RpcMsg rpcMsg = (RpcMsg) msg;

        switch (rpcMsg.getHead().getType()) {
            //  断连消息
            case TERMINATE: {
                ctx.channel().close();
                break;
            }
            //  ping消息
            case PING: {
                //  收到ping，回复pong
                RpcPongMsg pong = RpcPongMsg.build();
                pong.getHead().id(rpcMsg.getHead().getId());
                ctx.channel().writeAndFlush(pong);
                super.channelRead(ctx, msg);
                break;
            }
            //  业务消息
            case PUSH: {
                doPush(ctx, (RpcPushMsg) rpcMsg);
                super.channelRead(ctx, msg);
                break;
            }
            //  业务消息返回值
            case ACK: {
                doAck(ctx, (RpcAckMsg) rpcMsg);
                super.channelRead(ctx, msg);
                break;
            }
            //  未登录消息不予处理（PUSH消息的ACK，在未登录情况下，回来的就是这个消息）
            case LOGIN_NO: {
                futureExceptionByHead(ctx, rpcMsg.getHead(), new RpcNoLoginException("连接还未登录，服务端不予处理"));
                break;
            }
            //  登录成功 / 失败，直接走回调
            case LOGIN_SUCCESS: {
                futureByHead(ctx, rpcMsg.getHead(), true);
                super.channelRead(ctx, msg);
                break;
            }
            case LOGIN_ERROR: {
                futureExceptionByHead(ctx, rpcMsg.getHead(), new RpcLoginException("登录失败"));
                break;
            }
        }
    }


    /**
     *  执行push
     */
    protected void doPush(ChannelHandlerContext ctx, RpcPushMsg msg) {
    	//	在eventLoop线程反序列化
    	if (RpcConfigInfo.INSTANCE.isSerializeInPlace()) {
    		RpcPushBody body = SerializeThreadPool.INSTANCE_SERIALIZER.deSerialize(msg.getBodyBytes(), RpcPushBody.class);
    		//  执行过程在业务线程池中
    		serviceExecutor.invoke(body.getUrl(), body.getArgs(), body.getMetadata(), rer -> {
    			RpcAckMsg ack = RpcAckMsg.build();
    			ack.getHead().id(msg.getHead().getId());
    			if (!rer.isSuccess()) {
    				//  构建返回值
    				ack.body(RpcAckBody.build().success(false).errorMsg(rer.getE().getMessage()));
    				byte[] bytes = KryoThreadSerializer.INSTANCE.serialize(ack.getBody());
    				ctx.channel().writeAndFlush(ack.bodyBytes(bytes));
    			} else {
    				ack.body(RpcAckBody.build().success(true).returnValues(rer.getRtn()).metadata(rer.getMetadata()));
    				byte[] bytes = KryoThreadSerializer.INSTANCE.serialize(ack.getBody());
    				ctx.channel().writeAndFlush(ack.bodyBytes(bytes));
    			}
    		});
    	}
    	//	eventLoop -> serializeThread -> serviceThread -> serializeThread -> eventLoop
    	//	在序列化线程池中还原body
    	else {
        	SerializeThreadPool.INSTANCE.deSerlize(msg.getBodyBytes(), RpcPushBody.class, body -> {
	    		//  执行过程在业务线程池中
	    		serviceExecutor.invoke(body.getUrl(), body.getArgs(), body.getMetadata(), rer -> {
	    			RpcAckMsg ack = RpcAckMsg.build();
	    			ack.getHead().id(msg.getHead().getId());
	    			if (!rer.isSuccess()) {
	    				//  构建返回值
	    				ack.body(RpcAckBody.build().success(false).errorMsg(rer.getE().getMessage()));
	    				SerializeThreadPool.INSTANCE.serialize(ack.getBody(), bytes -> {
	    					ctx.channel().writeAndFlush(ack.bodyBytes(bytes));
	    				});
	    			} else {
	    				ack.body(RpcAckBody.build().success(true).returnValues(rer.getRtn()).metadata(rer.getMetadata()));
	    				SerializeThreadPool.INSTANCE.serialize(ack.getBody(), bytes -> {
	    					ctx.channel().writeAndFlush(ack.bodyBytes(bytes));
	    				});
	    			}
	    		});
	    	});
    	}
    }
    /**
     *  执行ack
     */
    @SuppressWarnings({ "rawtypes", "unchecked" })
	protected void doAck(ChannelHandlerContext ctx, RpcAckMsg msg) {
    	//	在eventLoop线程反序列化
    	if (RpcConfigInfo.INSTANCE.isSerializeInPlace()) {
    		RpcAckBody body = SerializeThreadPool.INSTANCE_SERIALIZER.deSerialize(msg.getBodyBytes(), RpcAckBody.class);
        	long id = msg.getHead().getId();
        	CompletableFuture cf = ctx.channel().attr(RpcAttribute.ackMapper).get().remove(id);
        	if (body.isSuccess()) {
        		cf.complete(body);
        	} else {
        		cf.completeExceptionally(new RpcRemoteException(body.getErrorMsg()));
        	}
    	}
    	//	在序列化线程池中还原body
    	else {
        	long id = msg.getHead().getId();
        	CompletableFuture cf = ctx.channel().attr(RpcAttribute.ackMapper).get().remove(id);
            //	在序列化线程池中还原body
            SerializeThreadPool.INSTANCE.deSerlize(msg.getBodyBytes(), RpcAckBody.class, body -> {
                //  执行过程在业务线程池中
                if (body.isSuccess()) {
                    cf.complete(body);
                } else {
                    cf.completeExceptionally(new RpcRemoteException(body.getErrorMsg()));
                }
          	});
    	}
    }


    /**
     * 解析消息头，并执行future
     */
    @SuppressWarnings({ "rawtypes", "unchecked" })
	protected void futureByHead(ChannelHandlerContext ctx, RpcMsgHead head, Object val) {
        long id = head.getId();
        CompletableFuture cf = ctx.channel().attr(RpcAttribute.ackMapper).get().remove(id);
        if (cf != null) {
            cf.complete(val);
        }
    }
    @SuppressWarnings("rawtypes")
	protected void futureExceptionByHead(ChannelHandlerContext ctx, RpcMsgHead head, Exception e) {
    	log.error("[RpcMsgReadHandler futureExceptionByHead] 接收到异常消息 head:{}", head, e);
    	
        long id = head.getId();
        CompletableFuture cf = ctx.channel().attr(RpcAttribute.ackMapper).get().remove(id);
        if (cf != null) {
            cf.completeExceptionally(e);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //  如果是读写相关异常，直接干掉连接
        if ((cause instanceof RpcReadException) || (cause instanceof RpcWriteException)) {
            log.error("读写相关异常，字节流已无法复原，直接断开连接重连", cause);
            ctx.channel().close().sync();
        }

        super.exceptionCaught(ctx, cause);
    }
}
